package jmine.tec.test.jmq.wmq;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.Queue;
import javax.jms.QueueConnection;
import javax.jms.QueueConnectionFactory;
import javax.jms.QueueReceiver;
import javax.jms.QueueSession;
import javax.jms.Session;
import javax.naming.Context;
import javax.naming.InitialContext;
import javax.naming.NamingException;

import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;

/**
 * Conector para ligar um messageListener a uma fila
 * 
 * @author takeshi
 */
public class JMSQueueConnector implements InitializingBean, DisposableBean {

    private QueueConnectionFactory queueConnectionFactory;

    private Destination destination;

    private MessageListener messageListener;

    private int threads = 1;

    private String destinationName;

    private ExecutorService executorService;

    private QueueConnection queueConnection;

    private Context jndiContext;

    private boolean useTemporaryQueue = true;

    private QueueSession queueSession;

    /**
     * {@inheritDoc}
     */
    public synchronized void afterPropertiesSet() throws Exception {
        this.executorService = Executors.newFixedThreadPool(this.threads);
        this.queueConnection = this.queueConnectionFactory.createQueueConnection();
        this.queueSession = this.queueConnection.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);
        QueueReceiver receiver = this.queueSession.createReceiver((Queue) this.getDestination());
        receiver.setMessageListener(new DistributionMessageListener());
        this.queueConnection.start();
    }

    /**
     * {@inheritDoc}
     */
    public void destroy() throws Exception {
        if (this.queueConnection != null) {
            this.queueConnection.close();
        }
        if (this.executorService != null) {
            this.executorService.shutdownNow();
        }
    }

    /**
     * @return the queueConnectionFactory
     */
    public QueueConnectionFactory getQueueConnectionFactory() {
        return this.queueConnectionFactory;
    }

    /**
     * @param queueConnectionFactory the queueConnectionFactory to set
     */
    public void setQueueConnectionFactory(QueueConnectionFactory queueConnectionFactory) {
        this.queueConnectionFactory = queueConnectionFactory;
    }

    /**
     * @return the messageListener
     */
    public MessageListener getMessageListener() {
        return this.messageListener;
    }

    /**
     * @param messageListener the messageListener to set
     */
    public void setMessageListener(MessageListener messageListener) {
        this.messageListener = messageListener;
    }

    /**
     * @return the threads
     */
    public int getThreads() {
        return this.threads;
    }

    /**
     * @param threads the threads to set
     */
    public void setThreads(int threads) {
        this.threads = threads;
    }

    /**
     * @return the destination
     */
    public String getDestinationName() {
        return this.destinationName;
    }

    /**
     * @param destinationName the destination to set
     */
    public void setDestinationName(String destinationName) {
        this.destinationName = destinationName;
    }

    /**
     * {@link MessageListener} que coloca a mensagem no thread pool para ser processado
     * 
     * @author takeshi
     */
    private class DistributionMessageListener implements MessageListener {

        /**
         * {@inheritDoc}
         */
        public void onMessage(Message arg0) {
            JMSQueueConnector.this.executorService.submit(new MessageHandlerTask(arg0));
        }

    }

    /**
     * Task para consumir uma mensagem
     * 
     * @author takeshi
     */
    private class MessageHandlerTask implements Runnable {
        private final Message message;

        /**
         * C'tor
         * 
         * @param message {@link Message}
         */
        public MessageHandlerTask(Message message) {
            this.message = message;
        }

        /**
         * {@inheritDoc}
         */
        public void run() {
            if (!Thread.interrupted()) {
                JMSQueueConnector.this.messageListener.onMessage(this.message);
            }
            Thread.currentThread().interrupt();
        }
    }

    /**
     * @return the destination
     * @throws JMSException e
     * @throws NamingException e
     */
    public Destination getDestination() throws JMSException, NamingException {
        if (this.destination == null) {
            if (this.isUseTemporaryQueue()) {
                this.destination = this.queueSession.createTemporaryQueue();
            } else {
                this.destination = (Destination) this.getJndiContext().lookup(this.getDestinationName());
            }
        }
        return this.destination;
    }

    /**
     * @param destination the destination to set
     */
    public void setDestination(Destination destination) {
        this.destination = destination;
    }

    /**
     * @param initialContext the initialContext to set
     */
    public void setJndiContext(Context initialContext) {
        this.jndiContext = initialContext;
    }

    /**
     * @return the useTemporaryQueue
     */
    public boolean isUseTemporaryQueue() {
        return this.useTemporaryQueue;
    }

    /**
     * @param useTemporaryQueue the useTemporaryQueue to set
     */
    public void setUseTemporaryQueue(boolean useTemporaryQueue) {
        this.useTemporaryQueue = useTemporaryQueue;
    }

    /**
     * @return the jndiContext
     * @throws NamingException e
     */
    public Context getJndiContext() throws NamingException {
        if (this.jndiContext == null) {
            this.jndiContext = new InitialContext();
        }
        return this.jndiContext;
    }

}
